package com.yanxu.customsink;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.FormattingMapper;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;

import java.util.HashMap;


/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2025-06-24-2025
 * @Description: 这里用一句话描述这个类的作用
 */
public class CustomSinkSample {
    public static void main(String[] args) throws Exception {
        //构建环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);

        //构建 source
        DataStreamSource<Event> dataStreamSource = environment.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L),
                new Event("Alice", "./prod?id=100", 3000L),
                new Event("Alice", "./prod?id=200", 3500L),
                new Event("Bob", "./prod?id=2", 2500L),
                new Event("Alice", "./prod?id=300", 3600L),
                new Event("Bob", "./home", 3000L),
                new Event("Bob", "./prod?id=1", 2300L),
                new Event("Bob", "./prod?id=3", 3300L)
        );
        /*new MapFunction<Event, String>(){
            @Override
            public String map(Event event) throws Exception {
                return "";
            }
        };*/

        /*SingleOutputStreamOperator<HashMap> map = dataStreamSource.map(event -> {
            String name = event.getName();
            String url = event.getUrl();
            HashMap hashMap = new HashMap();
            hashMap.put("name", name);
            hashMap.put("url", url);
            return hashMap;
        });*/

        // 使用 map 算子进行处理
        SingleOutputStreamOperator<String> map =
                dataStreamSource.map(new FormattingMapper<>(event -> event.getName()));

        //sink：使下沉（也就是下沉输出的位置）
        //map.print();
        //map.addSink(new PrintSinkFunction<>());
        map.addSink(new MySink());

        environment.execute();
    }
}